package com.dataWarehouseOss.rabbitmq.consumer;

import com.alibaba.fastjson.JSONObject;
import com.besttone.cryptography.tool.DecryptTool;
import com.dataWarehouseOss.rabbitmq.entity.DataEntity;
import com.dataWarehouseOss.rabbitmq.entity.IogItemEntity;
import com.dataWarehouseOss.rabbitmq.entity.Requestinfo;
import com.dataWarehouseOss.rabbitmq.entity.Responseinfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;

/**
 * @author ：LiuShihao
 * @date ：Created in 2020/8/17 9:03 上午
 * @desc ：kafak消费者
 */
@Component
@Slf4j
public class KafkaConsumer {
//    @Value("${ftp.ip}")
//    public  String FTP_IP ;
//    @Value("${ftp.ftp_port}")
//    public   Integer FTP_PORT ;
//    @Value("${ftp.username}")
//    public   String FTP_USERNAME ;
//    @Value("${ftp.password}")
//    public   String FTP_PASSWORD ;
//    @Value("${local.path}")
//    public   String LOCAL_PATH  ;
//    @Value("${ftp.ftpPath}")
//    public   String FTP_PATH ;

    public static FTPClient ftp;

    public static final String FTP_IP = "192.168.19.13";
    public static final Integer FTP_PORT = 21;
    public static final String FTP_USERNAME = "pengxiaoyang";
    public static final String FTP_PASSWORD = "pxy19840208";
    public static final String LOCAL_PATH = "/data/hw_data/";
    public static final String FTP_PATH = "/lshrabbitMQ/hw_data/";

    /**
     * 定义此消费者接收topics = "demo"的消息，与controller中的topic对应上即可
     * @param record 变量代表消息本身，可以通过ConsumerRecord<?,?>类型的record变量来打印接收的消息的各种信息
     */
    @KafkaListener(topics = "hw_data")
    public void listen (ConsumerRecord<String, String> record){
        log.info("============kafka Start================");
        log.info(FTP_IP+":"+FTP_PORT+"-"+FTP_USERNAME+"-"+FTP_PASSWORD+"-"+FTP_PATH+"-"+LOCAL_PATH);
        IogItemEntity iogItemEntity = JSONObject.parseObject(record.value(), IogItemEntity.class);
        Requestinfo requestinfo = JSONObject.parseObject(iogItemEntity.getRequestinfo(), Requestinfo.class);
        Responseinfo responseinfo = JSONObject.parseObject(iogItemEntity.getResponseinfo(), Responseinfo.class);
        List<DataEntity> dataEntities = JSONObject.parseArray(responseinfo.getData(), DataEntity.class);
        try {
            Assert.notEmpty(dataEntities,"异常信息：data为空，未查到数据！");
        }catch (IllegalArgumentException e){
            log.info(e.getMessage());
            dataEntities = new ArrayList<>();
            dataEntities.add(new DataEntity());
        }
        log.info(iogItemEntity.toString());
        log.info(requestinfo.toString());
        log.info(responseinfo.toString());
        DataEntity dataEntity = null;
        for (DataEntity dataEntity1 : dataEntities) {
            log.info(dataEntity1.toString());
            dataEntity = dataEntity1;
        }
        String NUMBER = "";
        try{
            Assert.notNull(requestinfo.getSubjectNum(),"异常信息：SubjectNum为null,名称使用ID");
            NUMBER = requestinfo.getSubjectNum();
        }catch (IllegalArgumentException e){
            log.info(e.getMessage());
            NUMBER = requestinfo.getId();
        }
        String encrpytSubjectNum = DecryptTool.datahouseDataEncrpyt(NUMBER);
        if (NUMBER.length()>=7){
            NUMBER = NUMBER.substring(0,7);
        }
        String isYN = "000000".equals(iogItemEntity.getResponsecode()) ? "Y":"N";
        String NAME = "";
        try{
            Assert.notNull(dataEntity.getName(),"异常信息：name为null,名称使用custName");
            NAME = dataEntity.getName();
        }catch (IllegalArgumentException e){
            log.info(e.getMessage());
            NAME = dataEntity.getCustName();
        }
        String encrpytName = DecryptTool.datahouseDataEncrpyt(NAME);
        String data = encrpytSubjectNum+"|"+NUMBER+"|"+iogItemEntity.getUserAgent()+"|"+iogItemEntity.getRequesttime()+"|"+isYN+"|"+encrpytName+"|"+dataEntity.getClassCode2()+"|"+dataEntity.getClassname2()+"\r\n";
//        log.info("顺序 :"+"号码加密|号码前7位|手机型号|调用时间|是否查到|名称加密|行业code|行业name");
        log.info("data:"+data);
        try {
            //获得ip地址
//            Set<String> inetAddresses = GetIPPortUtil.getInetAddresses();
//            String ip = "";
//            for (String inetAddress : inetAddresses) {
//                ip = inetAddress;
//            }
//            ip = "192.168.103.10";
//            String tomcatPort = GetIPPortUtil.getPortByMBean();
            byte[] body = data.getBytes();
            String today = DateTimeFormatter.ofPattern("yyyyMMdd").format(LocalDateTime.now());
            String hours = DateTimeFormatter.ofPattern("HH").format(LocalDateTime.now());
            //判断路径是否存在，如果不存在，则创建文件
            File file = new File(LOCAL_PATH);
            if (!file.exists()) {
                log.info("不存在该路径，开始创建...");
                file.mkdirs();
                log.info("路径创建完成...");
            }
//            file = new File(LOCAL_PATH+today+"_"+hours+"_"+ip+"_"+tomcatPort+".txt");
            file = new File(LOCAL_PATH+today+"_"+hours+".txt");
            if (!file.exists()) {
                log.info("不存在当天文件，开始创建...");
                file.createNewFile();
                log.info("文件创建完成...");
            }
            FileOutputStream fos = new FileOutputStream(file,true);
            fos.write(body);
            fos.close();
            System.out.println("---===开始连接ftp===---");
            ftp = new FTPClient();
            System.out.println("---===连接ftp===---");
            ftp.connect(FTP_IP, FTP_PORT);
            boolean login = ftp.login(FTP_USERNAME, FTP_PASSWORD);
            log.info("-----FTP登陆是否成功："+login);
//            FtpFileUtils.upload(file,FTP_PATH);
            boolean makeDirectory = ftp.makeDirectory(FTP_PATH);
            log.info("makeDirectory："+makeDirectory);
            ftp.changeWorkingDirectory(FTP_PATH);
            log.info("--开始上传ftp--");
            FileInputStream fis = new FileInputStream(file);
            boolean b = ftp.storeFile(file.getName(), fis);
            fis.close();
            log.info("--上传完成--"+b);
        } catch (IOException e) {
            e.printStackTrace();
            log.info("异常信息1："+e.getMessage());
        } catch (Exception e) {
            e.printStackTrace();
            log.info("异常信息2："+e.getMessage());
        }
        log.info("============kafka END================");
    }
}